package com.zhisheng.sql.blink.stream.example;

import com.zhisheng.common.utils.HttpUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.util.Date;

/**
 * CasinoWinJob
 *
 * @author yongxiang
 * @date 2025/2/28 14:59
 * @desc
 */
public class WinAlert {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        TableConfig config = tableEnv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled",true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay","5s");
        // 定义 Debezium Kafka 源表
        tableEnv.executeSql(
                "CREATE TABLE t_topology (\n" +
                        "  topology_id INT,\n" +
                        "  parent_topology_id INT,\n" +
                        "  name STRING,\n" +
                        "  status STRING,\n" +
                        "  topology_type_id INT\n" +
                        ")\n" +
                        "WITH\n" +
                        "  ('connector' = 'jdbc',\n" +
                        "    'url' = 'jdbc:mysql://192.168.50.235:3306/test_sm?serverTimezone=Asia/Shanghai',\n" +
                        "    'table-name' = 't_topology',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = 'Cyx1018!',\n" +
                        "    'sink.buffer-flush.max-rows' = '1', -- 实时写入（每条立即插入）\n" +
                        "    'sink.max-retries' = '3'\n" +
                        "  )");

        // 3. 创建过滤视图（WHERE id > 100）
        String createFilteredView =
                "CREATE TEMPORARY VIEW t_topology_view AS \n" +
                        "SELECT topology_id, parent_topology_id, name,topology_type_id \n" +
                        "FROM t_topology \n" +
                        "WHERE status = 'ACTIVE' AND topology_type_id IN (300,400,1200)";
        tableEnv.executeSql(createFilteredView);


        // 定义 Debezium Kafka 源表
        tableEnv.executeSql(
                "CREATE TABLE t_bet  ("
                        + "  bet_id INT,"
                        + "  wager DOUBLE,"
                        + "  player_id int,"
                        + "  casino_win DOUBLE, "
                        + "  table_id INT, "
                        + "  payout_complete_dtm BIGINT, " // 原始字段为 BIGINT（Unix 时间戳，毫秒）
                        + "  ts AS TO_TIMESTAMP_LTZ(payout_complete_dtm - 8 * 3600 * 1000, 3), "  // 转换为 TIMESTAMP
                        + "  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND"
                        + ") WITH ("
                        + "  'connector' = 'kafka',"
                        + "  'topic' = 'mysql-server.test_sm.t_bet',"
                        + "  'properties.bootstrap.servers' = '192.168.50.235:9092,192.168.50.235:9093,192.168.50.235:9094',"
                        + "  'properties.group.id' = 'win_alert',"
                        + "  'scan.startup.mode' = 'latest-offset',"
                        + "  'format' = 'debezium-json',"
                        + "  'debezium-json.schema-include' = 'true'"
                        + ")");


        Table joinedTable  = tableEnv.sqlQuery(
                        "  SELECT \n" +
                                "  tto.name AS table_name,\n" +
                                "  ttp.name AS pit_name, \n" +
                                "  tth.name AS hall,\n" +
                                "  tbu.wager, \n" +
                                "  tbu.casino_win," +
                                "  tbu.player_id\n" +
                                "FROM t_bet tbu\n" +
                                "LEFT JOIN t_topology_view tto \n" +
                                "  ON tbu.table_id = tto.topology_id\n" +
                                "LEFT JOIN t_topology_view ttp \n" +
                                "  ON tto.parent_topology_id = ttp.topology_id\n" +
                                "LEFT JOIN t_topology_view tth \n" +
                                "  ON ttp.parent_topology_id = tth.topology_id"
        );
        // 6. 转换为 DataStream
        DataStream<Row> resultStream = tableEnv.toChangelogStream(joinedTable);

        // 7. 实时告警处理
        DataStream<Alert> alertStream = resultStream
                .keyBy(row -> row.getFieldAs("player_id"))
                .process(new WinThresholdDetector())
                .returns(TypeInformation.of(Alert.class));


        // 4. 输出中间结果（调试用）
        Table betTable = tableEnv.from("t_bet");
        tableEnv.toChangelogStream(betTable).print("t_bet");
        // 8. 输出告警
        alertStream.addSink(new AlertSink());

        env.execute("Real-Time Casino Monitoring");



    }

    // 告警检测逻辑
    public static class WinThresholdDetector<K,T,I>
            extends KeyedProcessFunction<Integer, Row, Alert> {

        private final Double consoleAlert = 2000.00;

        private final Double emailAlert = 5000.00;

        private transient ValueState<Double> totalWinState;

        public WinThresholdDetector() {

        }


        @Override
        public void open(Configuration parameters) {
            ValueStateDescriptor<Double> descriptor =
                    new ValueStateDescriptor<>("totalWin", Double.class);
            totalWinState = getRuntimeContext().getState(descriptor);
        }

        @Override
        public void processElement(
                Row row,
                Context ctx,
                Collector<Alert> out) throws Exception {

            Double currentWin = (totalWinState.value() == null) ? 0.0 : totalWinState.value();

            Double newWin = currentWin.doubleValue() + (Double) row.getFieldAs("casino_win");

            // 更新状态
            totalWinState.update(newWin);

            if (newWin >= emailAlert){
                //发邮件
                sendEmail(new Alert(
                        ctx.getCurrentKey(),
                        newWin,
                        row.getFieldAs("table_name")));

            } else if (newWin >= consoleAlert) {
                out.collect(new Alert(
                        ctx.getCurrentKey(),
                        newWin,
                        row.getFieldAs("table_name")
                ));
            }

        }

        private void sendEmail(Alert alert) {
            System.out.println("send email...."+alert.toString());
            httpGet("http://127.0.0.1:5000/send?text=",alert);

        }

        public void httpGet(String urlStr,Alert alert) {
            try {
                String encode = URLEncoder.encode(alert.toString(), StandardCharsets.UTF_8.toString());
                URL url = new URL(urlStr+encode);
                HttpURLConnection conn = (HttpURLConnection) url.openConnection();
                conn.setRequestMethod("GET");

                // 设置超时和请求头
                conn.setConnectTimeout(5000);
                conn.setRequestProperty("User-Agent", "Java HTTP Client");

                int responseCode = conn.getResponseCode();
                if (responseCode == HttpURLConnection.HTTP_OK) {
                    try (BufferedReader in = new BufferedReader(
                            new InputStreamReader(conn.getInputStream()))) {
                        String inputLine;
                        StringBuilder response = new StringBuilder();
                        while ((inputLine = in.readLine()) != null) {
                            response.append(inputLine);
                        }
                        //System.out.println("Response: " + response.toString());
                    }
                } else {
                    System.out.println("GET request failed: " + responseCode);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    // 告警数据结构
    public static class Alert {
        public Integer playerId;
        public Double totalWin;
        public String tableName;

        // 必须包含空构造函数
        public Alert() {}

        public Alert(Integer playerId, Double totalWin, String tableName) {
            this.playerId = playerId;
            this.totalWin = totalWin;
            this.tableName = tableName;
        }

        @Override
        public String toString() {
            return String.format(
                    "Player %d won $%.2f at table %s [%s]",
                    playerId, totalWin, tableName, new Date()
            );
        }
    }

    // 告警输出Sink（示例）
    public static class AlertSink implements org.apache.flink.streaming.api.functions.sink.SinkFunction<Alert> {
        @Override
        public void invoke(Alert value, Context context) {
            System.err.println("ALERT: " + value);
            // 可扩展为写入Kafka/数据库/通知系统
        }
    }


}
